﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using ServiceManager.Services;
using ServiceManager.Utils;
using Protocol;

namespace QuotationService.RealtimeQuotation
{
	/// <summary>
	/// 行情消费者
	/// </summary>
	[DebuggerNonUserCode]
	public sealed class QuotationConsumer : IDisposable
	{
		/// <summary>
		/// 发布者
		/// </summary>
		Dictionary<QuotationSource, QuotationSubscribeService> Consumers = new Dictionary<QuotationSource, QuotationSubscribeService>();

        /// <summary>
        /// 基本资料订阅回调函数
        /// </summary>
		public event Action<QuotationSource, MKTSubscribeBasicInfoANS> OnBasicInfo;

        /// <summary>
		/// 行情订阅回调函数
		/// </summary>
		public event Action<QuotationSource, MKTSubscribeQuotationANS> OnQuotation;

		/// <summary>
		/// 市场信息订阅回调函数
		/// </summary>
		public event Action<QuotationSource, MKTSubscribeTradeStatusANS> OnMarketInfo;

		/// <summary>
		/// 
		/// </summary>
		public QuotationConsumer()
		{
		}

		/// <summary>
		/// 追加订阅通道
		/// </summary>
		/// <param name="channels"></param>
		public void AppendChannels(params QuotationSubscribeChannel[] channels)
		{
			lock (this)
			{
				foreach (var group in channels.GroupBy(x => x.Source))
				{
					if (!this.Consumers.ContainsKey(group.Key))
					{
                        var consumer = DirectoryService.GetQuotationSubscribeService(QuotServiceNames.PubServicePrefix + group.Key);
						consumer.Subscribe(group.Select(x => x.ToSubscribeChannel()).Distinct().ToArray());
						this.Consumers[group.Key] = consumer;
						consumer.RTSHandler = new Action<string, string, string, byte[]>(Consumer_OnMessage);
					}
					else
					{
						this.Consumers[group.Key].Subscribe(group.Select(x => x.ToSubscribeChannel()).Distinct().ToArray());
					}
				}
			}
		}

		/// <summary>
		/// 移除订阅通道（解订阅）
		/// </summary>
		/// <param name="channels"></param>
		public void RemoveChannels(params QuotationSubscribeChannel[] channels)
		{
			lock (this)
			{
				foreach (var group in channels.GroupBy(x => x.Source))
				{
					if (this.Consumers.ContainsKey(group.Key))
					{
						this.Consumers[group.Key].UnSubscribe(group.Select(x => x.ToSubscribeChannel()).Distinct().ToArray());
					}
				}
			}
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="service"></param>
		/// <param name="channel"></param>
		/// <param name="match"></param>
		/// <param name="message"></param>
		void Consumer_OnMessage(string service, string channel, string match, byte[] message)
		{
			//////lock (this)
			{
				//订阅通道
				QuotationChannel qc = new QuotationChannel(channel);

				var pkg = new DMP.Package();

				switch (qc.DataType)
				{
					//市场信息
					case QuotationDataType.MarketInfo:
						if (this.OnMarketInfo != null)
						{
							if (pkg.ReadFromBuffer(message, 0) > 0)
								this.OnMarketInfo(qc.Source, pkg.Message as MKTSubscribeTradeStatusANS);
						}
						break;


					case QuotationDataType.BasicInfo:
						if (this.OnBasicInfo != null)
						{
							if (pkg.ReadFromBuffer(message, 0) > 0)
								this.OnBasicInfo(qc.Source, pkg.Message as MKTSubscribeBasicInfoANS);
						}
						break;

					case QuotationDataType.Quotation:
						if (this.OnQuotation != null)
						{
							if (pkg.ReadFromBuffer(message, 0) > 0)
								this.OnQuotation(qc.Source, pkg.Message as MKTSubscribeQuotationANS);
						}
						break;
				}
			}
		}

		/// <summary>
		/// 
		/// </summary>
		public void Dispose()
		{
			lock (this)
			{
				this.Consumers.Values.ToList().ForEach(x => x.Dispose());
				this.Consumers.Clear();
			}
		}
	}
}
